
Last time we introduced the trivial namesake Timestamped interface, and used the excellent ArrayDeque of Java6 to collect such things, imposing a time-based capacity and some external synchronization. Now let's test this with some threads.

<!--break-->

<p>   
<a style='text-decoration: none;' href="https://code.google.com/p/vellum/wiki/home">
<div style="border-bottom: solid 1px; background-color: #ccff00;">
<img src="http://upload.wikimedia.org/wikipedia/commons/1/12/Gnome-appointment-new.svg" border="0" align="left" hspace="0"/><span style="font-style: normal; font-weight: bold; font-size: 12pt;">&nbsp;Deque shredder:</span><span style="font-style: italic; font-weight: normal; font-size: 11pt;">&nbsp;A part of "Timestamped: a trilogy in a few parts."</span>
</div>
</a>

In case you missed the update to our <a href="http://weblogs.java.net/blog/evanx/archive/2012/06/27/timestamped-deque">previous installment</a>, our generic <tt>TimestampedElement</tt> now has a <tt>compareTo()</tt> for <a href="http://docs.oracle.com/javase/6/docs/api/java/lang/Comparable.html">natural ordering</a> by the timestamp. 

<pre>
public class TimestampedElement<T> implements Timestamped, Comparable<Timestamped> {
    T element;
    long timestamp;

    public TimestampedElement(T element, long timestamp) {
        this.element = element;
        this.timestamp = timestamp;
    }

    public T getElement() {
        return element;
    }
    
    @Override
    public long getTimestamp() {
        return timestamp;
    }

    @Override
    public int compareTo(Timestamped other) {
        if (timestamp < other.getTimestamp()) return -1;
        if (timestamp > other.getTimestamp()) return 1;
        else return 0;
    }    
}
</pre>

For completeness herewithin, our <a href="http://weblogs.java.net/blog/evanx/archive/2012/06/27/timestamped-deque">deque collector</a> for our <tt>Timestamped</tt> thingies, follows.

<pre>
public class TimestampedDequer<T extends Timestamped>  {
    long capacityMillis;
    long lastTimestamp;
    ArrayDeque<T> deque = new ArrayDeque();
    
    public TimestampedDequer(long capacityMillis) {
        this.capacityMillis = capacityMillis;
    }
    
    public synchronized void addLast(T element) {
        if (element.getTimestamp() == 0 || element.getTimestamp() < lastTimestamp) {
            deque.clear(); // throw our toys out the cot exception
        } else {
            lastTimestamp = element.getTimestamp();
            prune(lastTimestamp);
            deque.addLast(element);
        }
    }

    private void prune(long latestTimestamp) {
        while (deque.size() > 0 && 
                deque.getFirst().getTimestamp() <= latestTimestamp - capacityMillis) {
            deque.removeFirst();
        }
    }

    public synchronized Deque<T> snapshot(long lastTimestamp) {
        prune(lastTimestamp);
        return deque.clone();
    }

    public synchronized Deque<T> tail(int size) {
        Deque tail = new ArrayDeque();
        Iterator<T> it = deque.descendingIterator();
        for (int i = 0; i < size && it.hasNext(); i++) {
            tail.addFirst(it.next());
        }
        return tail;
    }    
}
</pre>
<img src="http://jroller.com/evanx/resource/bicycle-guardians-deck-rider-black-front-crop.jpg" align="right" hspace="8"/>
where we use the efficient <a href="http://docs.oracle.com/javase/6/docs/api/java/util/ArrayDeque.html"><tt>ArrayDeque</tt></a> implementation of Java6.

As discussed last time, we remove expired elements from the head when we add the latest element to the tail, to make it self-pruning. And we provide a synchronized <tt>snapshot()</tt> and <tt>tail()</tt> for a couple of use-cases as follows... 

Decisively, we will use <tt>snapshot()</tt> to analyse the latest records for the desired interval e.g. for an automated status check every minute.

Furthermore, we will use a size-based <tt>tail()</tt> e.g. to display the latest so-many records in a status web page, for informational purposes.

Now let's do us some "heavy-dropping" with threads, using a <tt>ScheduledExecutorService</tt> to regularly add records to the deque. 
<pre>
public class TimestampedDequerTest  {
    long capacityMillis = 90;
    long scheduledInterval = 10;
    long scheduledDelay = 0;
    final TimestampedDequer<TimestampedElement> dequer = new TimestampedDequer(capacityMillis);
    boolean verbose = false;
    
    Runnable scheduledRunnable = new Runnable() {

        @Override
        public void run() {
            addLast();
        }
    };

    private void addLast() {
        long timestamp = System.currentTimeMillis();
        String value = "record at " + timestamp;
        dequer.addLast(new TimestampedElement(value, timestamp));
        if (verbose) {
            System.out.println(value);
        }
    }

    @Test
    public void testConcurrently() throws Exception {
        ScheduledExecutorService executorService = Executors.newScheduledThreadPool(capacity);
        ScheduledFuture future = executorService.scheduleAtFixedRate(
                    scheduledRunnable, scheduledDelay, scheduledInterval, TimeUnit.MILLISECONDS);
        checkConcurrently();
        checkConcurrently();
        future.cancel(true);
    }
</pre>
where we check twice, just to make sure! ;) Actually we want to make sure of the <tt>prune()</tt>'ing, following the sleep for <tt>capacityMillis</tt>.
<pre>    
    private void checkConcurrently() throws Exception {
        long startMillis = System.currentTimeMillis();
        System.out.println("startMillis " + startMillis);
        verbose = true;
        Thread.sleep(capacityMillis);
        int expectedCapacity = (int) (capacityMillis / scheduledInterval);
        verbose = false;
        long stopMillis = System.currentTimeMillis();
        System.out.println("stopMillis " + stopMillis);
        Deque<TimestampedElement> deque = dequer.snapshot(stopMillis);
        long firstTimestamp = deque.getFirst().getTimestamp();   
        long lastTimestamp = deque.getLast().getTimestamp();   
        System.out.println("size " + deque.size());
        System.out.println("first " + firstTimestamp);
        System.out.println("last " + lastTimestamp);
        Assert.assertTrue("first time", firstTimestamp >= startMillis);        
        Assert.assertTrue("last time", lastTimestamp >= firstTimestamp);
        Assert.assertTrue("capacityMillis min", lastTimestamp - firstTimestamp >= 0);        
        Assert.assertTrue("capacityMillis max", lastTimestamp - firstTimestamp <= capacityMillis);        
        Assert.assertTrue("size min", deque.size() > 0);
        Assert.assertTrue("size max", deque.size() <= expectedCapacity);
        checkSet(deque);
    }    
</pre>
which prints...
<pre>
scheduledInterval 10
record at 1340231378158
record at 1340231378168
record at 1340231378178
...
record at 1340231378228
record at 1340231378238
startMillis 1340231378157
stopMillis 1340231378247
size 9
first 1340231378158
last 1340231378238
...
startMillis 1340231378249
stopMillis 1340231378339
size 9
first 1340231378258
last 1340231378338
</pre>
We survey this output, eyeing the timestamps, and nod ponderously.

Just for good measure, we add the records to a <tt>SortedSet</tt>, and check that the first and last timestamps match. 
<pre>
    private void checkSet(Deque<TimestampedElement> deque) throws Exception {
        SortedSet<Timestamped> set = new TreeSet();
        set.addAll(deque);        
        Assert.assertEquals("size", deque.size(), set.size());
        Assert.assertEquals("first", deque.getFirst().getTimestamp(), set.first().getTimestamp());
        Assert.assertEquals("last", deque.getLast().getTimestamp(), set.last().getTimestamp());
    }
</pre>
where our <tt>TimestampedElement</tt>'s <tt>compareTo()</tt> method enables the natural ordering, and forgoes having to use our <tt>TimestampedComparator</tt> to construct the <tt>SortedSet</tt>.

Let's vary the <tt>scheduledInterval</tt>. 
<pre>
    @Test
    public void testScheduledIntervals() throws Exception {
        while (--scheduledInterval > 0) {
            ScheduledFuture future = Executors.newScheduledThreadPool(10).scheduleAtFixedRate(
                    scheduledRunnable, scheduledDelay, scheduledInterval, TimeUnit.MILLISECONDS);
            Thread.sleep(capacityMillis);
            int expectedCapacity = (int) (capacityMillis / scheduledInterval);
            long stopMillis = System.currentTimeMillis();
            Deque<TimestampedElement> deque = dequer.snapshot(stopMillis);
            Woohoo.assertEquals("interval " + scheduledInterval, expectedCapacity, deque.size());
            future.cancel(true);
            Thread.sleep(scheduledInterval);
        }
    }
</pre>
where we loop the <tt>scheduledInterval</tt> down to 1ms.
<pre>
interval 9: Woohoo! 10 == 10
interval 8: Woohoo! 11 == 11
interval 7: Woohoo! 12 == 12
interval 6: D'oh! 15 != 14
interval 5: D'oh! 18 != 17
interval 4: Woohoo! 22 == 22
interval 3: D'oh! 30 != 29
interval 2: D'oh! 45 != 44
interval 1: D'oh! 90 != 89
</pre>
Given how unpredictable time is, ironically, with those threads and what-not, we can't exactly predict the size of the list. D'oh! So for that we have used the following util class to see if the size is more or less what we expect...
<pre>
public class Woohoo {

    public static void assertEquals(String message, Object expected, Object actual) {
        if (actual.equals(expected)) {
            System.out.printf("%s: Woohoo! %s == %s\n", message, expected, actual);
        } else {
            System.out.printf("%s: D'oh! %s != %s\n", message, expected, actual);
        }
    }
</pre>
Selectively using the above drop-in replacement for <tt>Assert</tt>, we get our tests to pass 100%, woohoo! ;)

To further increase the intensity, we allow the scheduled threads to coalesce.
<pre>
    @Test
    public void testCumulativeScheduledIntervals() throws Exception {
        Deque<ScheduledFuture> futures = new ArrayDeque();
        int expectedCapacity = 0;
        while (--scheduledInterval > 0) {
            expectedCapacity += (int) (capacityMillis / scheduledInterval);
            ScheduledFuture future = Executors.newScheduledThreadPool(100).scheduleAtFixedRate(
                    scheduledRunnable, scheduledDelay, scheduledInterval, TimeUnit.MILLISECONDS);
            futures.add(future);
            Threads.sleep(capacityMillis);
            long stopMillis = System.currentTimeMillis();
            Deque<TimestampedElement> deque = dequer.snapshot(stopMillis);
            Woohoo.assertEquals("interval " + scheduledInterval, expectedCapacity, deque.size());
            Threads.sleep(scheduledInterval);
        }
        for (ScheduledFuture future : futures) {
            future.cancel(true);
        }
    }
</pre>
where we only cancel all the scheduled threads when all is said and done. As expected, we get some D'oh's and Woohoo's!
<pre>
interval 9: Woohoo! 10 == 10
interval 8: Woohoo! 21 == 21
interval 7: D'oh! 33 != 34
interval 6: Woohoo! 48 == 48
interval 5: Woohoo! 66 == 66
interval 4: D'oh! 88 != 89
interval 3: Woohoo! 118 == 118
interval 2: D'oh! 163 != 165
interval 1: D'oh! 253 != 254
</pre>

Let's wrap this up by asserting that our <tt>Deque</tt> collector is reasonably robust!? 

Next time, we'll use our timestamped deque to capture log4j records, and as alluded to previously, regularly analyse the latest <tt>deque</tt> of logs to detect when our app's wheels are wobbling or even coming off altogether, and notify ourselves thereof.

<h3>Resources</h3>

https://code.google.com/p/vellum/ - where i will collate these articles and their code.


